package com.mq.client;

import com.rabbitmq.client.*;

import java.io.IOException;

/**
 * 消息持久化
 *
 * @author xwj
 * @date 2019/2/28
 */
public class PersistenceWorker {
    private final static String QUEUE_NAME = "hello";

    public static void main(String[] argv) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        // 开启持久化
        boolean durable = true;
        //一次只接受一条未包装的消息（见下文）,实现公平分发消息
        int prefetchCount = 1;
        channel.basicQos(prefetchCount);
        channel.queueDeclare(QUEUE_NAME, durable, false, false, null);
        System.out.println("Waiting for task");
        final Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
                    throws IOException {
                String task = new String(body, "UTF-8");
                System.out.println("Received task:" + task);
                try {
                    doWork(task);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    System.out.println("Work Done");
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            }
        };
        // 开启返回消息确认，默认关闭，手动设置false打开
        boolean autoAck = false;
        channel.basicConsume(QUEUE_NAME, autoAck, consumer);
    }

    private static void doWork(String task) throws InterruptedException {
        System.out.println("Working");
        for (char ch : task.toCharArray()) {
            if (ch == '.') {
                Thread.sleep(1000);
            }
        }
    }
}
